package job.mfp.document_builder;

import java.io.IOException;
import java.util.ArrayList;
import java.util.EnumMap;
import java.util.Iterator;
import java.util.List;

import mapred.base.TextReducer;
import mapred.example.model.Fields;

import org.apache.hadoop.io.Text;

public class MFP_DocumentBuilder_Reducer extends TextReducer {

    @Override
    public void reduce(Text key, Iterator<Text> values) throws IOException {
        EnumMap<Fields, List<String>> map = new EnumMap<Fields, List<String>>(Fields.class);
        
        for (Fields f : Fields.values()) {
            map.put(f, new ArrayList<String>());
        }
        
        while (values.hasNext()) {
            String value = values.next().toString();
            int pos = value.indexOf("\t");
            Fields tag = Fields.valueOf(value.substring(0, pos));
            map.get(tag).add(value);
        }
        
        if (map.get(Fields.URL_SEG).size() > 0) {
            this.output(new Text("Document_Begin"), new Text(""));
            this.output(key, new Text(map.get(Fields.URL_SEG).get(0)));
            this.output(key, new Text(map.get(Fields.TITLE).get(0)));
            this.output(key, new Text(map.get(Fields.BODY).get(0)));
            for (String anchor : map.get(Fields.ANCHOR)) {
                this.output(key, new Text(anchor));
            }
            this.output(new Text("Document_End"), new Text(""));
        }
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
